#include "pipe_module.h"
#include "pipe_session.h"
#include "default_interface.hpp"
#include "time_wheel.h"
#include "main_service.h"

namespace anet {
	namespace pipe {
		const int gTimeWheelSize = 1023;
		// implementation GetPipeModule().
		IPipeModule* GetPipeModule() {
			return &CPipeModule::instance();
		}

		CPipeModule::CPipeModule() : m_loop(1), 
			m_server(m_loop), m_pipeReporter(nullptr), 
			m_timeWheel(gTimeWheelSize) {
		}

		CPipeModule::~CPipeModule() {
			m_server.stop();
		}

		CPipeModule& CPipeModule::instance() {
			static CPipeModule gInstance;
			return gInstance;
		}

		bool CPipeModule::Run(int count) {
			bool ret = false;
			// anet run.
			if (anet::tcp::CMainService::instance().run(count)) {
				ret = true;
			}

			// timer run.
			m_timeWheel.run();

			return ret;
		}

		uint32 CPipeModule::GetLocalId() const {
			return m_pipeConfig.GetServerId();
		}

		bool CPipeModule::Init(const char* pipePath, IPipeReporter* reporter) {
			if (reporter == nullptr) {
				return false;
			}

			if (!m_pipeConfig.Load(pipePath)) {
				return false;
			}
			return this->start(reporter);
		}

		IPipeReporter* CPipeModule::GetReporter() {
			return m_pipeReporter;
		}

		userSessionFactory* CPipeModule::GetSessionFactory() {
			return &m_sessionFactory;
		}

		bool CPipeModule::CheckConnectInfo(unsigned int remoteId, const std::string& token) const {
			// remote id check.
			if (this->GetLocalId() != remoteId) {
				return false;
			}

			// token check.
			for (size_t i = 0;i < m_pipeConfig.GetListens().size();i++) {
				if (m_pipeConfig.GetListens()[i].token == token) {
					return true;
				}
			}
			return false;
		}

		bool CPipeModule::start(IPipeReporter* reporter) {
			if (reporter == nullptr) {
				return false;
			}

			m_pipeReporter = reporter;
			if (!this->initPipeDown(m_pipeConfig.GetListens())) {
				return false;
			}

			if (!this->initPipeUp(m_pipeConfig.GetConnects())) {
				return false;
			}
			return true;
		}

		bool CPipeModule::initPipeDown(const std::vector<listenInfo>& listens) {
			m_server.setPacketParser(&m_codec);
			m_server.setSessionFactory(&m_sessionFactory);
			for (size_t i = 0;i < listens.size();i++) {
				if (!m_server.start(listens[i].addr, listens[i].port)) {
					return false;
				}
			}
			return true;
		}

		bool CPipeModule::initPipeUp(const std::vector<connectInfo>& connects) {
			for (size_t i = 0;i < connects.size();i++) {
				auto session = m_sessionFactory.createSession();
				if (session == nullptr) {
					LogAdebug("create session return nullptr");
					return false;
				}

				auto pSharePtrSession = (sharePtrSession*)session;
				assert(pSharePtrSession != nullptr && "pipe session is nullptr");

				CPipeSession *pipeSession = pSharePtrSession->getProxySession();
				assert(pipeSession != nullptr && "pipe session is nullptr");

				// initialize the pipe's session.
				pipeSession->SetRemoteId(connects[i].serverId);
				pipeSession->SetToken(connects[i].token);
				pipeSession->SetMode(ePipeMode::pipe_mode_connector);

				// create client object.
				anet::tcp::CClient* client = new anet::tcp::CClient(m_loop);
				pipeSession->SetClient(client);

				// initialize client's codec and session.
				client->setPacketParser(&m_codec);
				client->setSession(session);
				if (!client->asyncConnect(connects[i].addr, connects[i].port, 
					// reconnect to the server if there is an error.
					[client](const char* ip, unsigned short port, int err) {
					  (void)ip;(void)port;(void)err;
					  // try to reconnect the remote server again.
					  // pSharePtrSession->Release();
					  client->reAsyncConnect();
				})) {
					return false;
				}
			}
			return true;
		}

		STimeWheelSpace::CTimeWheel& CPipeModule::getTimeWheel() {
			return m_timeWheel;
		}
	}
}
